import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.linalg.Vector;
import org.apache.spark.mllib.linalg.Vectors;
import org.apache.spark.rdd.RDD;

/**
 * @description Spark-DBSCAN算子测试类
 * @date 2021-12-23
 */
public class DBSCANDemo {

    public static void main(String[] args) throws Exception {

        String filePath = "D:\\sales.csv";
        String savePath = "D:\\tmpsave2.txt";

        SparkConf conf = new SparkConf()
                .setAppName("ay_test")
                .setMaster("local[*]")
                .set("spark.executor.memory", "4g")
                .set("spark.driver.memory", "4g");

        JavaSparkContext javaSparkContext = new JavaSparkContext(conf);

        JavaRDD<String> lines = javaSparkContext.textFile(filePath);
        RDD<Vector> vectorRDD = lines.map(row -> {
            String[] tmps = row.split(",");
            double[] values = new double[4];
            for (int i = 1; i < 3; ++i) {
                values[i - 1] = Double.parseDouble(tmps[i]);
            }
            return Vectors.dense(values);
        }).rdd();

//        DBSCAN dbscan = DBSCAN.train(
//            vectorRDD,
//            0.5,
//            5,
//            10
//        );
//        JavaRDD<String> map = dbscan.labeledPoints().toJavaRDD()
//            .map(p -> String.format("%s,%s,%s", p.x(), p.y(), p.cluster()));
//        System.out.println(map.take(10));
//        javaSparkContext.stop();

//        DbscanSettings dbscanSettings = new DbscanSettings()
//            .withEpsilon(0.5)
//            .withNumberOfPoints(10);
//
//        DbscanModel model = Dbscan.train(vectorRDD, dbscanSettings);
    }

}
